package big.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Collection;
import java.util.Properties;

//手动维护offset
public class CustomOffsetConsumer {

    public static void main(String[] args) {
        Properties po = new Properties();
        po.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop129:9092");
        po.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        po.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        po.put(ConsumerConfig.GROUP_ID_CONFIG,"test"); //消费者组id  只要此id相同就属于同一个消费者组
        po.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");  //关闭自动提交 默认为true
        //创建一个消费者
        final KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(po);
        consumer.subscribe(Arrays.asList("first"), new ConsumerRebalanceListener() {
            //提交当前负债分区的offset
            public void onPartitionsRevoked(Collection<TopicPartition> topicPartitions) {
                System.out.println("=================回收分区===================");
                for (TopicPartition topicPartition :
                        topicPartitions) {
                    System.out.println("topicPartion: "+ topicPartition);
                }
            }

            //定位新分配分区的offset
            public void onPartitionsAssigned(Collection<TopicPartition> topicPartitions) {
                System.out.println("=================分配分区===================");
                for (TopicPartition topicPartition :
                        topicPartitions) {
                    System.out.println("topicPartion: "+ topicPartition);
                    long offset = getTopicPartionOffset(topicPartition);
                    //consumer.seek(topicPartition,offset);  //定义当前的offset
                }
            }
        }); //订阅的主题
        //调用poll 到分区拉取数据
        while(true){
            ConsumerRecords<String, String> consumerRecords = consumer.poll(100); //每一百秒到分区拉取一次数据
            for (ConsumerRecord<String, String> consumerRecord:consumerRecords){
                System.out.printf("offset = %d, partion = %s, value = %s%n", consumerRecord.offset(), consumerRecord.partition(), consumerRecord.value());
                TopicPartition topicPartition = new TopicPartition(consumerRecord.topic(), consumerRecord.partition());
                commitOffset(topicPartition,consumerRecord.offset()+1);

            }
        }

    }

    private static void commitOffset(TopicPartition topicPartition, long l) {
        //自定义提交offset
    }

    private static Long getTopicPartionOffset(TopicPartition topicPartition) {
        //获取offset的方法,一般将offset放到缓存中
        return null;
    }

}
